package com.shujia.stream

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo3DriverHA {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("stream")
    //.setMaster("local[2]") //指定两个线程

    val sc = new SparkContext(conf)

    val chechpoint = "/data/driver/ha"

    val createStreamingContext = () => {

      println("创建StreamingContext.................................")

      //创建sparkstreaming上下文对象
      val ssc = new StreamingContext(sc, Durations.seconds(5))

      //自动减Driver中的信息保存到hdfs
      ssc.checkpoint(chechpoint)

      val ds = ssc.socketTextStream("192.168.129.101", 8888)
      ds
        .flatMap(_.split(","))
        .map((_, 1))
        .reduceByKey(_ + _)
        .print()

      ssc
    }

    //如果checkpoint里面有StreamingContext的信息，直接基于checkpoint里面的数据创建一个StreamingContext
    //如果没有，调用后面的方法创建StreamingContext
    val ssc = StreamingContext.getOrCreate(chechpoint, createStreamingContext)


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }
}
